package cm.kafka.controller;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
public class ProducerController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/KafkaSendMsg")
    public String data(@RequestBody JSONObject jsonObject,
                       @RequestParam(value = "topic") String topic,
                       @RequestParam(value = "num") Integer num,
                       @RequestParam(value = "n") Integer n){
        log.info("-------------测试开始-------------");
        long time, t;
        for (int j = 0; j < n; j++) {
            time = System.currentTimeMillis();
            // 通过kafka发送出去
            for (int i = 0; i < num; i++) {
                kafkaTemplate.send(topic, JSONObject.toJSONString(jsonObject));
            }

            t = System.currentTimeMillis() - time;
            log.info("当前到达发送次数：" + (j+1) +"发送条数："+num+ ",发送耗时：" + t + "ms");
            if (1000L - t > 0) {
                try {
                    Thread.sleep(1000L - t);
                    log.info("等待了" + (1000L - t) + "ms，发送下一条数据");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                log.info("条数太多无法处理,本次耗时：" + t + "ms");
            }
        }
        log.info("-------------测试结束-------------");
        return "ok";
    }

}
